services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:90

• Initial onboarding materializes every commit ID and retains every parsed DB/Kafka record until the entire archive finishes. For LKML-scale archives this is unbounded memory usage and no progress is checkpointed before a likely OOM/restart. Process bounded commit batches and persist each batch/head checkpoint before continuing, similar to the 250-commit chunking in git_integration/src/crowdgit/services/commit/commit_service.py:709-743.
  .github/workflows/backend-lint.yaml:147
• This CI job installs pytest and the PR adds a substantial test suite, but the job runs only Ruff. As a result, parser regressions can merge even though the PR reports the tests as coverage. Run pytest in this job after lint/format checks.

      - name: Check Python linting and formatting
        if: steps.changes.outputs.python_changed == 'true'
        run: |
          uv run ruff check src/ --output-format=github
          uv run ruff format --check src/

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:31

• This introduces a new class-based worker, contrary to the repository's functions-over-classes direction for new service code (CLAUDE.md:42-43 and services-checklist.md:41-43). Keep shutdown state in a small functional runner/context and expose plain processing functions so the polling and per-list logic remain independently testable.

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:84

• This ignores the stored source_url; ensure_mirror always clones https://lore.kernel.org/{name}/. A valid public-inbox URL with a different host or path will silently ingest the wrong archive. Pass the source URL to the mirror layer and keep the local directory key separate.
  services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:91
• On initial onboarding, new_commits(..., None) materializes the shard’s entire history, and this method then retains every DB and Kafka record until all shards finish. Public-inbox shards can contain hundreds of thousands of messages, so the worker can exhaust memory before reaching the existing insert chunks. Iterate in bounded batches and checkpoint each batch only after persistence and emission succeed.
  services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:93
• read_email launches synchronous Git subprocesses on the FastAPI event loop—twice per message. A large import will block health requests, shutdown handling, and other async work for long periods. Offload this blocking call to a thread.
  services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:94
• Any parser exception escapes this per-commit loop, marks the whole list failed, and leaves the shard head unchanged. The next retry starts from the same malformed message, so one poison email can permanently block every later message in the shard. Catch failures per commit and skip/quarantine them with observable logging or metrics, as the Git integration does in commit_service.py:652-655.
  services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:72
• This helper is called by the long-running worker, so sys.exit() raises SystemExit, bypasses its except Exception handlers, and can terminate the service because of one malformed commit. Raise a domain exception and let the worker apply its per-message failure policy instead.
  services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:87
• The blob-read failure also calls sys.exit(). In worker mode this can terminate the entire service instead of failing one message/list. Raise a normal domain exception here, consistent with the commit-not-found branch.
  services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:354
• Messages without Message-ID are emitted with an empty sourceId. Activity deduplication keys include sourceId, so multiple such messages in the same segment/platform/type/channel can collapse into one activity. Use a stable fallback such as the Git commit/blob ID, or explicitly skip these messages.
  services/apps/mailing_list_integration/src/crowdmail/services/mirror/mirror_service.py:58
• The five-minute default applies to initial public-inbox-clone and fetch operations. LKML-scale archives can legitimately take much longer, so onboarding will repeatedly kill the clone and never complete. Use a separately configurable mirror timeout sized for large archives while retaining shorter limits for lightweight commands.
  services/apps/mailing_list_integration/src/crowdmail/database/crud.py:117
• A process crash after acquisition leaves state='processing' and lockedAt set. This exclusion, combined with both selectors requiring lockedAt IS NULL, means the list is never acquired again; stale onboarding rows also consume the concurrency count forever. Treat locks as leases and reclaim/reset processing rows older than a configured threshold.

    states_to_exclude = (ListState.PENDING, ListState.PROCESSING, ListState.PENDING_REONBOARD)

services/apps/mailing_list_integration/README.md:16

• This documentation still says the new worker emits platform=groupsio, while the implementation, seed, activity type, and identities all use platform=mailinglist. The same stale Groups.io references recur at lines 46, 62, and 66; update them so operators verify the actual platform.

  parses new messages, writes activities to `integration.results`, and emits
  Kafka messages to the `data-sink-worker` topic — same plumbing as
  git_integration, with `platform=groupsio`.

services/libs/data-access-layer/src/mailinglist/index.ts:128

• This reset misses the integration-deletion path. IntegrationService.destroyAll() soft-deletes the integration but does not soft-delete its mailinglist.lists rows; the ownership check then permits a new integration to claim that URL. Because e."deletedAt" is still null, the upsert changes the owner while preserving the old owner's shard heads, so the new project permanently skips all previously processed history. Include the previous integrationId (or deleted owner state) in existing and reset processing whenever ownership changes.
  services/apps/mailing_list_integration/src/crowdmail/database/crud.py:180
• This counts every historical integration result, not the pending backlog described by the method. Once the table contains 5,000,000 processed/error rows, onboarding is disabled forever even with zero queued work. Restrict the bounded scan to pending rows (the table has an index on state).

        integration_results_count = await fetchval(
            "SELECT COUNT(*) FROM (SELECT 1 FROM integration.results LIMIT $1) t",
            (MAX_INTEGRATION_RESULTS,),
        )

services/apps/mailing_list_integration/src/crowdmail/database/crud.py:89

• The onboarding cap is not atomic across worker replicas. For example, with a count of 2 and a cap of 3, two transactions can both observe c.count = 2, lock different rows via SKIP LOCKED, and both transition them to processing, producing 4 concurrent onboardings. Serialize the count-and-claim operation with a database semaphore/advisory lock acquired before the statement, or model a fixed number of claimable slots.

            AND l."deletedAt" IS NULL
            AND i."deletedAt" IS NULL
            AND c.count < $3
        ORDER BY lp.priority ASC, lp."createdAt" ASC
        LIMIT 1
        FOR UPDATE OF lp SKIP LOCKED

backend/src/api/integration/helpers/mailingListAuthenticate.ts:32

• Requiring HTTPS and blocking only literal loopback names is not an SSRF boundary. Inputs such as https://10.0.0.1/..., private/link-local IPv6, localhost., or a public hostname that resolves or redirects to an internal address all pass and are later fetched by public-inbox-clone from the worker network. Use an explicit host allowlist or resolve and reject all non-public addresses at validation time and again for every worker-side redirect; also reject URL credentials.

const isSafeSourceUrl = (sourceUrl: string): boolean => {
  try {
    const url = new URL(sourceUrl)
    return url.protocol === 'https:' && !isBlockedHost(url.hostname.toLowerCase())

backend/src/api/integration/helpers/mailingListAuthenticate.ts:45

• This only removes trailing slashes, so equivalent URLs such as https://LORE.kernel.org/list and https://lore.kernel.org:443/list remain distinct. That bypasses duplicate detection, the ownership check, advisory locking, and the database uniqueness constraint, allowing the same archive to be connected to multiple projects and ingested twice. Canonicalize with URL semantics (host casing, default port, normalized path) and explicitly reject credentials, query strings, and fragments before deduplication and storage.

// "https://host/list" and "https://host/list/" must resolve to the same DB
// row — the worker's ensure_mirror() already normalizes to this form before
// cloning (mirror_service.py), so storage must match or the same archive
// ends up mirrored twice under two rows. Kept as a plain function (not a
// zod .transform()/.preprocess()) since either makes the field optional in
// z.infer with the installed zod v4 — reproduced in isolation, unrelated to
// this schema's nesting.
export const canonicalizeSourceUrl = (sourceUrl: string): string => sourceUrl.replace(/\/+$/, '')

services/libs/data-access-layer/src/mailinglist/index.ts:112

• Reassigning the existing row to a new integration is unsafe while a worker may still hold this list ID. Worker checkpoint/state writes are keyed only by listId, so an old owner's in-flight run can overwrite the reset after this transaction commits, advance the new owner's heads, and cause its historical messages to be skipped. Use a new row or an ownership/lease generation that every worker write must match.
  services/apps/mailing_list_integration/src/crowdmail/database/crud.py:82
• A legitimate onboarding run that lasts beyond this timeout is reclaimed even though the original worker is still alive, because lockedAt is set only on acquisition and never refreshed. The two workers can then write the same shard checkpoint, and JSONB merge does not prevent the older worker from regressing that shard key. Add a renewable lease/owner token and make checkpoint/final-state writes conditional on it.

                lp.state = $1
                AND lp."lastProcessedAt" IS NULL
                AND lp."lockedAt" < NOW() - INTERVAL '1 hour' * $4::numeric
            )

services/apps/mailing_list_integration/src/crowdmail/database/crud.py:135

• The recurrent stale-reclaim path can also acquire a still-running worker: no code refreshes lockedAt during processing. Runs exceeding four hours may overlap, duplicate Kafka work, and regress a same-shard checkpoint when the slower worker writes later. Use a renewable lease/owner token and reject stale-owner updates.

                lp.state = $1
                AND lp."lastProcessedAt" IS NOT NULL
                AND lp."lockedAt" < NOW() - INTERVAL '1 hour' * $6::numeric
            )

services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:237

• An empty or whitespace-only From: header is not None, so parse_author returns an empty verified username. The data sink filters both empty identities and rejects the activity, permanently dropping the message after this worker advances its head. Apply the same per-commit synthetic author fallback when parsing yields an empty address.
  backend/src/api/integration/helpers/mailingListAuthenticate.ts:32
• This validation still permits SSRF targets such as RFC1918/ULA/link-local addresses, internal DNS names, and redirects to private services; https is not a network-boundary check. Because the worker passes this value to public-inbox-clone, validate the resolved destination (and every redirect) at fetch time or restrict sources to an explicit public-inbox allowlist.

const isSafeSourceUrl = (sourceUrl: string): boolean => {
  try {
    const url = new URL(sourceUrl)
    return (

backend/src/api/integration/helpers/mailingListAuthenticate.ts:37

• Requiring HTTPS and blocking only loopback literals still permits targets such as https://10.0.0.1, https://169.254.169.254, IPv6 private/link-local addresses, public DNS names resolving privately, and redirects to those hosts. Because the worker passes this tenant-controlled URL to public-inbox-clone, the endpoint can issue requests into the internal network. Enforce an approved-host policy or resolve and reject every non-public address on every redirect, backed by network egress restrictions.

      url.protocol === 'https:' &&
      !isBlockedHost(url.hostname.toLowerCase()) &&
      // Credentials/query/fragment have no meaning for a public-inbox archive
      // URL; rejecting them keeps canonicalizeSourceUrl a lossless
      // normalization instead of one that silently drops caller-supplied data.

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:116

• This branch logs that the commit will be retried but then falls through to state = COMPLETED; mark_list_processed sets lastProcessedAt, so the failed commit is not retried until the normal 24-hour recurrent interval. Re-raise the operational read failure so the list is marked FAILED and uses the shorter failure retry interval.
  services/apps/data_sink_worker/src/service/member.service.ts:162
• When this conflict is caused by an identity the member already owns with verified=false, the case-insensitive missing filter drops an incoming verified=true version and never promotes the stored row. The call then succeeds while the verified assertion is silently lost. As in syncIdentitiesAfterRedirect, update matching unverified identities before retrying the genuinely missing ones.

        const missing = deduped.filter(
          (incoming) =>
            !current.some(
              (existing) =>
                existing.platform === incoming.platform &&

services/apps/data_sink_worker/src/service/activity.service.ts:1845

• This redirect path also treats an existing (platform, type, value) as complete regardless of verified. If the owner has that identity unverified and the incoming activity asserts it verified, it is excluded from missingIdentities, so the verification upgrade is silently discarded. Promote matching unverified rows before inserting the missing identities, as MemberService.syncIdentitiesAfterRedirect already does.

            !ownerIdentities.some(
              (existing) =>
                existing.platform === incoming.platform &&
                existing.type === incoming.type &&
                existing.value.trim().toLowerCase() === incoming.value.trim().toLowerCase(),

services/apps/mailing_list_integration/src/crowdmail/database/crud.py:89

• The concurrency cap is not atomic across worker processes: concurrent transactions can both read the same count below the limit, lock different listProcessing rows via SKIP LOCKED, and each transition one to PROCESSING. This can exceed MAX_CONCURRENT_ONBOARDINGS precisely during expensive initial clones. Serialize the count-and-acquire decision (for example with a transaction advisory lock) or use a database-backed semaphore.

            AND c.count < $3
        ORDER BY lp.priority ASC, lp."createdAt" ASC
        LIMIT 1
        FOR UPDATE OF lp SKIP LOCKED

backend/src/api/integration/helpers/mailingListAuthenticate.ts:27

• This host check is bypassable: Node's URL parser reports IPv6 literals with brackets, so https://[::1]/... does not equal ::1; private/link-local addresses and hostnames resolving to them are also accepted. Because the worker passes this tenant-controlled URL to public-inbox-clone, this leaves an SSRF path into the worker's network. Resolve and reject all non-public addresses (including redirects), or restrict sources to an explicit public-inbox allowlist, and enforce the check again at fetch time.

const isSafeSourceUrl = (sourceUrl: string): boolean => {

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:116

• This break turns a failed read into a successful run: control later sets the list state to COMPLETED and advances lastProcessedAt, even though this shard still has unread commits. Re-raise the operational error so the list is marked FAILED and uses the failed retry cadence; any unflushed commits remain safely uncheckpointed for retry.
  services/apps/data_sink_worker/src/service/activity.service.ts:1862
• insertMemberIdentities defaults to failOnConflict=false, so this recovery path silently DO NOTHINGs any missing identity that is already owned by another member. The activity is then attached to ownerId without merging or surfacing the second identity conflict, leaving identities split across members. Route these identities through the same conflict-aware redirect/merge logic used by MemberService.syncIdentitiesAfterRedirect instead of using a conflict-ignoring insert.

          await insertMemberIdentities(
            this.pgQx,
            missingIdentities.map((identity) => ({
              memberId: ownerId,
              platform: identity.platform,
              value: identity.value,
              type: identity.type,
              verified: identity.verified,
              source: identity.source,
              sourceId: identity.sourceId,
              integrationId: identity.integrationId,
              verifiedBy: identity.verifiedBy,
            })),
          )

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:87

• The core ingestion/checkpoint path has no tests; the added suite only exercises noteren.py. Please cover DB-insert → Kafka-send → head-update ordering, partial flushes, read failures, parse/date skips, and state/lock finalization. This is especially important because the analogous git ingestion path has integration coverage in services/apps/git_integration/src/test/test_activity_extraction.py.

backend/src/api/integration/helpers/mailingListAuthenticate.ts:25

• This host check still permits SSRF. An authenticated caller can supply RFC1918, link-local, or private IPv6 destinations (and URL.hostname represents IPv6 loopback as [::1], which this comparison misses), while a public hostname can resolve or redirect to a private address. Because the worker passes this URL to public-inbox-clone, it can be made to access internal HTTPS services. Restrict sources to an allowlist or validate resolved addresses and every redirect again in the worker before connecting.

const isBlockedHost = (h: string): boolean =>
  h === 'localhost' || h === '::1' || h === '0.0.0.0' || h.startsWith('127.')

services/apps/mailing_list_integration/src/crowdmail/database/crud.py:234

• Merging JSONB does not prevent a stale worker from moving a shard checkpoint backward: lockedAt is never refreshed during processing, so stale-reclaim can start a second worker, and both workers overwrite the same shard key here. If the slower worker flushes last, subsequent polls re-read and requeue the already processed range. Add lease heartbeats plus a fencing token/conditional checkpoint update, or otherwise ensure only the current lease holder can advance heads.

    sql_query = """
    UPDATE mailinglist."listProcessing"
        SET "lastProcessedHeads" = "lastProcessedHeads" || $1::jsonb,
        "updatedAt" = NOW()
    WHERE "listId" = $2

services/apps/data_sink_worker/src/service/member.service.ts:165

• This reconciliation treats an existing identity as complete without considering verification. If a concurrent write created the identity as unverified while this payload carries verified: true, the retry silently drops the promotion. The redirect path below already promotes this exact transition; do the same here before retrying missing identities.

        const currentMap = await findIdentitiesForMembers(this.pgQx, [memberId])
        const current = currentMap.get(memberId) ?? []
        const missing = deduped.filter(
          (incoming) =>
            !current.some(
              (existing) =>
                existing.platform === incoming.platform &&
                existing.type === incoming.type &&
                existing.value.trim().toLowerCase() === incoming.value.trim().toLowerCase(),
            ),

services/apps/data_sink_worker/src/service/activity.service.ts:1856

• This second identity-sync implementation also ignores verification state. When the owner already has the same identity as unverified and the incoming payload verifies it, missingIdentities excludes it and the owner remains unverified. Reuse the redirect synchronization logic that both inserts missing identities and promotes verified=false to true, rather than duplicating only half of it here.

        const ownerIdentities =
          (await findIdentitiesForMembers(this.pgQx, [ownerId])).get(ownerId) ?? []
        const missingIdentities = incomingIdentities.filter(
          (incoming) =>
            !ownerIdentities.some(
              (existing) =>
                existing.platform === incoming.platform &&
                existing.type === incoming.type &&
                existing.value.trim().toLowerCase() === incoming.value.trim().toLowerCase(),
            ),

services/apps/mailing_list_integration/src/crowdmail/server.py:22

• The PR description says FastAPI imports and lifespan behavior are tested, but the only added pytest module is test_noteren.py and it never imports this server or invokes lifespan. Add a lifespan test that verifies worker startup and graceful shutdown (including timeout cancellation), or correct the stated test coverage.